package study.java.jdk.jdk8._1_stream.parallel;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.RandomUtil;
import com.google.common.base.Stopwatch;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * （1）ParallelStreams 默认使用 ForkJoinPool.commonPool()线程池
 * 注意：默认情况下，你写的 ParallelStreams 都是通过该线程池调度执行，整个应用程序都共享这个线程池。
 */
@Slf4j
public class _1_parallel_Test {

    private List<String> dataLt = Arrays.asList("1001", "1002", "1003");

    @Test
    public void test1() {
        Stopwatch stopwatch = Stopwatch.createStarted();

        //
        List<String> userLt = dataLt.parallelStream()
                .map(e -> {
                    int sec = RandomUtil.randomInt(1, 5);
                    log.info(">>>>>> sleep {} sec", sec);
                    ThreadUtil.sleep(sec * 1000);
                    return e + "---";
                })
                .collect(Collectors.toList());
        log.info(">>>>>> {}", userLt);

        log.info(">>>>>> cost time [{}] s", stopwatch.elapsed(TimeUnit.SECONDS));
    }

    @SneakyThrows
    @Test
    public void test2() {
        Stopwatch stopwatch = Stopwatch.createStarted();

        //
        ForkJoinPool pool = new ForkJoinPool(1);
        List<String> userLt = pool.submit(() -> dataLt.parallelStream()
                        .map(e -> {
                            int sec = RandomUtil.randomInt(1, 5);
                            log.info(">>>>>> sleep {} sec", sec);
                            ThreadUtil.sleep(sec * 1000);
                            return e + "---";
                        })
                        .collect(Collectors.toList()))
                .get();
        pool.shutdown();
        log.info(">>>>>> {}", userLt);

        log.info(">>>>>> cost time [{}] s", stopwatch.elapsed(TimeUnit.SECONDS));
    }

    @SneakyThrows
    @Test
    public void test3() {
        Stopwatch stopwatch = Stopwatch.createStarted();

        //
        int size = 1;
        ThreadPoolExecutor pool = new ThreadPoolExecutor(size,
                size, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1));
        List<String> userLt = pool.submit(() -> dataLt.parallelStream()
                .map(e -> {
                    int sec = RandomUtil.randomInt(1, 5);
                    log.info(">>>>>> sleep {} sec", sec);
                    ThreadUtil.sleep(sec * 1000);
                    return e + "---";
                })
                .collect(Collectors.toList())).get();
        pool.shutdown();
        log.info(">>>>>> {}", userLt);

        log.info(">>>>>> cost time [{}] s", stopwatch.elapsed(TimeUnit.SECONDS));
    }
}
